﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Collections.Concurrent;
using DataTransFrame;
using System.Net;
using System.Net.Sockets;
using NetMQ;
using NetMQ.Sockets;

namespace DataTramsmittingServer
{
   public class DataTransServer
    {
        

        #region 服务端处理

        public void TransServerThread(object obj)
        {
            //int[] ports ={ 38401, 38400 };
            //if (obj != null)
            //{
            //    int[] Arr = (int[])obj;
            //    ports[0] = Arr[0];
            //    ports[1] = Arr[1];
            //}
            //信息发布服务
            ThreadPool.QueueUserWorkItem(new WaitCallback(ConsumeMsg), 38401);
            //创建服务器
            TcpListener server = new TcpListener(IPAddress.Parse("0.0.0.0"), 38400);
            server.Start();
            while (true)
                ThreadPool.QueueUserWorkItem(new WaitCallback(CreateMsg), server.AcceptTcpClient());

        }
        private void CreateMsg(object obj)
        {
            try
            {
                using (TcpClient client = obj as TcpClient)
                {
                    NetworkStream ns = client.GetStream();
                    if (client.Client.Poll(-1, SelectMode.SelectRead))
                    {
                        System.Threading.Thread.Sleep(10);
                        if (client.Available > 10)
                        {
                            byte[] bytes = new byte[client.Available];
                            ns.Read(bytes, 0, bytes.Length);
                            cqc.Enqueue(new Common().GetProtobufObjectFromBytes<DataTransFrame.CollectToTranServer>(bytes));
                        }
                    }
                }
            }
            catch (Exception ex) { Console.WriteLine(ex.ToString()); }
        }
        private ConcurrentQueue<DataTransFrame.CollectToTranServer> cqc = new ConcurrentQueue<CollectToTranServer>();
        private void ConsumeMsg(object obj)
        {
            while (true)
            {
                try
                {
                    using (PublisherSocket pub = new PublisherSocket())
                    {
                        pub.Bind($"tcp://*:{obj}");
                        CollectToTranServer cts = null;
                        while (true)
                        {
                            if (cqc.Count == 0)
                                System.Threading.Thread.Sleep(500);

                            if (cqc.TryDequeue(out cts))
                            {
                                TranServerToTerminal tst = new TranServerToTerminal() { TOPIC = cts.TOPIC, CMD = cts.CMD, CHECKTIME = cts.CHECKTIME, RECORDS = cts.RECORDS };
                                byte[] bytes = new Common().GetBytesFromProtoObject(tst);

                                pub.SendMoreFrame(cts.IDCard).SendFrame(bytes, bytes.Length);
                                System.Threading.Thread.Sleep(1);
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.ToString());
                }
            }
        }

        #endregion
    }
}
